package SparkMLlib.SparkMLlibItemAls

import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.{SparkConf, SparkContext}
/**
  * 协同过滤最小二乘法demo，基于用户推荐：根据用户的相似度来为某个用户推荐物品
  */
object UserCFDemo {

  /**
    * 解析String，获取Rating
    * @param str
    * @return
    */
  def parseRating(str:String):Rating={
    val fields = str.split("\t")
    Rating(fields(0).toInt,fields(1).toInt,fields(2).toDouble)
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("UserCFDemo").setMaster("local[*]")
    val sc = new SparkContext(conf)
    //用户评价数据：用户ID  影片ID  星级  时间戳
    val ratingData = sc.textFile("D://GitProjects//SparkKafkaHadoopZookeeperHBaseHiveRedis//spark//src//main//resource//movie.data")
    //读取数据，生成RDD并转换成Rating对象
    val ratingsRDD = ratingData.map(parseRating(_))
    //隐藏因子数(也即隐藏特征，理论讲因子数越多效果越好，但太多也会造成训练模型和保存时所需的内存开销，所以一般可取50~200之间)
    val rank=50
    //最大迭代次数(每次迭代都能降低评价矩阵的重建误差，但一般经少数次迭代后ALS模型已能收敛成一个比较合理的好模型)
    val maxIter=10
    //正则化因子(该参数控制模型的正则化过程，从而控制模型的过拟合情况)
    val lambda=0.01
    //训练模型
    val model = ALS.train(ratingsRDD,rank,maxIter,lambda)

    //从电影ID到标题的映射
    val movies = sc.textFile("E:\\test\\ml-100k\\u.item")
    val titles = movies.map(line=>line.split("\\|")).map(array=>(array(0).toInt,array(1))).collectAsMap()

    //推荐物品（电影）数量
    val K=10
    //用户1
    val user1=66
    //推荐结果
    val topKRecs = model.recommendProducts(user1,K)
    println("用户"+user1)
    topKRecs.foreach(rec=>{
      val movie = titles(rec.product)
      val rating = rec.rating
      println(s"推荐电影：$movie ，预测评分：$rating")
    })
  }
}

